package com.examples.redis.reader;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.util.Preconditions;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisSentinelPool;

import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

public class Flink2RedisReader  extends RichSourceFunction< String[]> implements Cloneable{
    private static final long serialVersionUID = 3334654984032091675L;
    private JedisPool jedisPool;
    private JedisSentinelPool jedisSentinelPool;
    private boolean running=false;
    private int keyLength = 0;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        FlinkJedisPoolConfig flinkJedisConfigBase = new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build();
        GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
        genericObjectPoolConfig.setMaxIdle(flinkJedisConfigBase.getMaxIdle());
        genericObjectPoolConfig.setMaxTotal(flinkJedisConfigBase.getMaxTotal());
        genericObjectPoolConfig.setMinIdle(flinkJedisConfigBase.getMinIdle());
        JedisPool jedisPool = new JedisPool(genericObjectPoolConfig, flinkJedisConfigBase.getHost(), flinkJedisConfigBase.getPort(), flinkJedisConfigBase.getConnectionTimeout(), flinkJedisConfigBase.getPassword(), flinkJedisConfigBase.getDatabase());
        this.jedisPool = jedisPool;
        this.jedisSentinelPool = null;
        running = true;
    }

    @Override
    public void run(SourceContext< String[]> sourceContext) throws Exception {
        while(running){
            Jedis jedis = null;
            String[] smembers = null;
            try {
                jedis = this.getInstance();
                String key = "spider_results:30";
                smembers = jedis.smembers(key).toArray(new String[]{});
            } catch (Exception e) {
                throw e;
            } finally {
                this.releaseInstance(jedis);
            }
            if(keyLength != smembers.length){
                sourceContext.collect(smembers);
                keyLength = smembers.length;
            }
        }
    }

    @Override
    public void cancel() {

    }

    private Jedis getInstance() {
        return this.jedisSentinelPool != null ? this.jedisSentinelPool.getResource() : this.jedisPool.getResource();
    }

    private void releaseInstance(Jedis jedis) {
        if (jedis != null) {
            try {
                jedis.close();
            } catch (Exception ignored) {}

        }
    }
}
